package com.lgj.rocketmq.code;

import com.alibaba.fastjson.JSONObject;
import com.lgj.rocketmq.config.ConsumerProperties;
import com.lgj.rocketmq.message.RocketMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.ReflectionUtils;

import java.lang.reflect.Method;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Slf4j
public class MQPushConsumerFactory implements InitializingBean {
    private ConsumerProperties consumerProperties;

    private ConsumerInitFactory consumerInitFactory;

    public MQPushConsumerFactory(ConsumerProperties consumerProperties,ConsumerInitFactory consumerInitFactory){
        this.consumerInitFactory = consumerInitFactory;
        this.consumerProperties = consumerProperties;
    }

    private static final Map<String,DefaultMQPushConsumer> MQPushConsumerMap = new HashMap<>();

    public Map<String,DefaultMQPushConsumer> getMQPushConsumers(){
        return MQPushConsumerMap;
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        Map<String,Consumer> consumerMap =  consumerInitFactory.getConsumers();
        consumerMap.forEach((topic,consumer)->{
            MQPushConsumerMap.put(topic,initMQPushConsumer(consumer));
        });
    }

    private DefaultMQPushConsumer initMQPushConsumer(Consumer consumer){
        DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer();
        defaultMQPushConsumer.setNamesrvAddr(consumerProperties.getNameServerAddress());
        defaultMQPushConsumer.setConsumerGroup(consumer.getGroup());

        if (consumer.getMaxReconsumeTimes() != null) {
            defaultMQPushConsumer.setConsumeThreadMax(consumer.getMaxReconsumeTimes());
        }
        if (consumer.getMaxReconsumeTimes()== null && consumerProperties.getMaxReconsumeTimes() != null){
            defaultMQPushConsumer.setConsumeThreadMax(consumerProperties.getMaxReconsumeTimes());
        }
        if (consumerProperties.getConsumeThreadMax() != null){
            defaultMQPushConsumer.setConsumeThreadMax(consumerProperties.getConsumeThreadMax());
        }
        try {
            defaultMQPushConsumer.subscribe(consumer.getTopic(), (String) null);
        } catch (Exception e) {
            log.error("消费者【{}】初始化失败，错误信息：{}", JSONObject.toJSONString(consumer),e.getMessage());
        }
        defaultMQPushConsumer.setMessageModel(consumer.getMessageModel());

        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                try{
                    for (MessageExt message : msgs) {
                        RocketMessage rocketMessage  = JSONObject.parseObject(message.getBody(), RocketMessage.class);
                        log.info("收到消息【{}】，",JSONObject.toJSONString(rocketMessage));
                        String topic = consumer.getTopic();
                        Method handeMethod = consumerInitFactory.getHandeMessageMethods().get(topic);
                        Object handBean = consumerInitFactory.handeMessageBeanMap().get(topic);
                        ReflectionUtils.invokeMethod(handeMethod,handBean,rocketMessage);
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }catch (Exception e){
                    log.error("消费消息失败，消息内容【{}】，错误信息：{}",JSONObject.toJSONString(msgs),e.getMessage());
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            }
        });
        return defaultMQPushConsumer;
    }
}
